/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.binlog.comsumer.server.listener;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.fasterxml.jackson.core.type.TypeReference;
import com.iwindplus.base.domain.exception.BizException;
import com.iwindplus.base.util.JacksonUtil;
import com.iwindplus.binlog.comsumer.server.domain.dto.BinlogRowDataDTO;
import com.iwindplus.binlog.comsumer.server.domain.dto.SourceMetaDTO;
import com.iwindplus.binlog.comsumer.server.domain.property.BinLogConsumerProperty;
import com.iwindplus.binlog.comsumer.server.handler.BinlogActionHandler;
import jakarta.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

/**
 * binlog消费方日志监听器.
 *
 * @author zengdegui
 * @since 2025/08/24 13:04
 */
@Slf4j
public class BinLogConsumerListener {

    @Resource
    protected BinLogConsumerProperty property;

    @Resource
    private BinlogActionHandler handler;

    @KafkaListener(
        topics = {"${spring.kafka.configs[0].topic-name}"},
        groupId = "${spring.kafka.configs[0].group-id}"
    )
    public void listenSingle(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        AtomicInteger counter = new AtomicInteger(0);

        records.stream()
            .filter(Objects::nonNull)
            .map(ConsumerRecord::value)
            .filter(CharSequenceUtil::isNotBlank)
            .map(this::parseData)
            .filter(CollUtil::isNotEmpty)
            .flatMap(List::stream)
            .peek(dto -> counter.incrementAndGet())
            .forEach(this::processData);

        log.info("本次共解析 {} 条 binlog 行数据", counter.get());

        if (ack != null) {
            ack.acknowledge();
        }
    }

    private void processData(BinlogRowDataDTO entity) {
        try {
            this.handler.processHandler(entity);
        } catch (Exception ex) {
            if (ex instanceof BizException bizEx) {
                log.warn("binlog日志消息消费失败，错误信息={}", bizEx.getBizMessage());
                return;
            }

            log.error("binlog日志消息消费失败={}", ex);
        }
    }

    private List<BinlogRowDataDTO> parseData(String value) {
        // log.info("binlog日志消息消费参数={}", value);
        if (CharSequenceUtil.isBlank(value)) {
            return null;
        }

        final BinlogRowDataDTO data = JacksonUtil.parseObject(value, new TypeReference<>() {
        });

        if (Objects.isNull(data)) {
            return null;
        }

        if (!checkTableNeedSign(data)) {
            return null;
        }

        return List.of(data);
    }

    private boolean checkTableNeedSign(BinlogRowDataDTO data) {
        final SourceMetaDTO source = data.getSource();
        if (Objects.isNull(source)) {
            return false;
        }
        final String db = source.getDb();
        final String table = source.getTable();
        if (CharSequenceUtil.isBlank(db) || CharSequenceUtil.isBlank(table)) {
            return false;
        }
        // 判断是否是需要验签的表，不需要则跳过
        if (!property.checkNeedSign(db, table)) {
            return false;
        }

        return true;
    }
}
